package com.ksyun.campus.metaserver.services;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.BlockInfo;
import com.ksyun.campus.metaserver.domain.FileType;
import com.ksyun.campus.metaserver.domain.Inode;
import com.ksyun.campus.metaserver.domain.MetaServerInfo;
import com.ksyun.campus.metaserver.storage.MetadataStore;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 元数据同步服务
 * 负责 Leader-Follower 之间的数据同步
 */
@Slf4j
@Service
public class MetadataSyncService {
    
    @Autowired
    private CuratorFramework curator;
    
    @Autowired
    private MinfsProperties properties;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Autowired
    private LeaderElectionService leaderElection;
    
    @Autowired
    private MetadataStore metadataStore;
    
    private RestTemplate restTemplate;
    private ExecutorService syncExecutor;
    private CuratorCache syncEventCache;
    
    // 同步相关路径
    private String syncEventsPath;
    private String syncStatusPath;
    
    // 同步状态
    private volatile boolean isInitialized = false;
    private volatile boolean isMetadataStoreReady = false;
    private final AtomicLong lastSyncSequence = new AtomicLong(0);
    private final ConcurrentHashMap<String, SyncStatus> followerSyncStatus = new ConcurrentHashMap<>();
    
    // 同步事件类型
    public enum SyncEventType {
        FILE_CREATE,
        FILE_DELETE,
        DIRECTORY_CREATE,
        DIRECTORY_DELETE,
        METADATA_UPDATE,
        BLOCK_ALLOCATE,
        BLOCK_COMMIT
    }
    
    // 同步事件
    public static class SyncEvent {
        public long sequence;
        public SyncEventType type;
        public String path;
        public Map<String, Object> data;
        public long timestamp;
        public String leaderId;
        
        // Constructors, getters, setters
        public SyncEvent() {}
        
        public SyncEvent(SyncEventType type, String path, Map<String, Object> data) {
            this.type = type;
            this.path = path;
            this.data = data;
            this.timestamp = System.currentTimeMillis();
        }
    }
    
    // 同步状态
    public static class SyncStatus {
        public String instanceId;
        public long lastSyncSequence;
        public LocalDateTime lastSyncTime;
        public boolean isSyncing;
        public String syncError;
        public long totalInodes;
        public long totalBlocks;
        public String dataChecksum;
        
        public SyncStatus() {}
        
        public SyncStatus(String instanceId) {
            this.instanceId = instanceId;
            this.lastSyncSequence = 0;
            this.lastSyncTime = LocalDateTime.now();
            this.isSyncing = false;
        }
    }
    
    // 数据一致性检查结果
    public static class ConsistencyCheckResult {
        public boolean isConsistent;
        public Map<String, InconsistencyInfo> inconsistencies;
        public LocalDateTime checkTime;
        public String checkId;
        
        public ConsistencyCheckResult() {
            this.inconsistencies = new HashMap<>();
            this.checkTime = LocalDateTime.now();
            this.checkId = java.util.UUID.randomUUID().toString();
        }
    }
    
    // 不一致信息
    public static class InconsistencyInfo {
        public String type; // EXTRA_DATA, MISSING_DATA, DATA_MISMATCH
        public String description;
        public Map<String, Object> details;
        public String suggestedAction;
        
        public InconsistencyInfo(String type, String description, String suggestedAction) {
            this.type = type;
            this.description = description;
            this.suggestedAction = suggestedAction;
            this.details = new HashMap<>();
        }
    }
    
    @PostConstruct
    public void initialize() throws Exception {
        log.info("初始化元数据同步服务");
        
        // 初始化路径
        String rootPath = properties.getZookeeper().getRootPath();
        syncEventsPath = rootPath + "/meta/sync/events";
        syncStatusPath = rootPath + "/meta/sync/status";
        
        // 创建基础路径
        ensureBasePaths();
        
        // 清理可能存在的无效同步事件数据
        cleanupInvalidSyncEvents();
        
        // 初始化 REST 客户端
        restTemplate = new RestTemplate();
        
        // 初始化线程池
        syncExecutor = Executors.newFixedThreadPool(3);
        
        // 添加 Leader 选举监听器
        leaderElection.addLeadershipListener(this::handleLeadershipChange);
        
        // 启动同步事件监听器
        startSyncEventListener();
        
        isInitialized = true;
        log.info("元数据同步服务初始化完成");
    }
    
    @PreDestroy
    public void shutdown() throws Exception {
        log.info("关闭元数据同步服务");
        
        isInitialized = false;
        
        if (syncEventCache != null) {
            syncEventCache.close();
        }
        
        if (syncExecutor != null) {
            syncExecutor.shutdown();
            syncExecutor.awaitTermination(30, TimeUnit.SECONDS);
        }
        
        log.info("元数据同步服务已关闭");
    }
    
    /**
     * 标记元数据存储已准备好，可以开始处理同步事件
     */
    public void markMetadataStoreReady() {
        this.isMetadataStoreReady = true;
        log.info("元数据存储已准备好，开始处理同步事件");
    }
    
    private void ensureBasePaths() throws Exception {
        String[] paths = {
                syncEventsPath,
                syncStatusPath
        };
        
        for (String path : paths) {
            if (curator.checkExists().forPath(path) == null) {
                curator.create()
                        .creatingParentsIfNeeded()
                        .forPath(path);
                log.debug("创建同步路径: {}", path);
            }
        }
    }
    
    private void handleLeadershipChange(boolean isLeader) {
        try {
            if (isLeader) {
                log.info("成为 Leader，启动数据同步协调");
                startLeaderSyncCoordination();
            } else {
                log.info("成为 Follower，启动数据同步接收");
                startFollowerSyncReceiver();
            }
        } catch (Exception e) {
            log.error("处理 Leader 身份变化时出错", e);
        }
    }
    
    private void startLeaderSyncCoordination() {
        // Leader 负责协调同步
        syncExecutor.submit(() -> {
            while (leaderElection.isLeader() && isInitialized) {
                try {
                    // 检查 Follower 同步状态
                    checkFollowerSyncStatus();
                    
                    // 清理过期的同步事件
                    cleanupOldSyncEvents();
                    
                    Thread.sleep(5000); // 5秒检查一次
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("Leader 同步协调出错", e);
                }
            }
        });
    }
    
    private void startFollowerSyncReceiver() {
        // Follower 定期从 Leader 拉取同步数据
        syncExecutor.submit(() -> {
            while (!leaderElection.isLeader() && isInitialized) {
                try {
                    // 从 Leader 拉取增量同步数据
                    pullSyncEventsFromLeader();
                    
                    Thread.sleep(2000); // 2秒同步一次
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("Follower 同步接收出错", e);
                }
            }
        });
    }
    
    private void startSyncEventListener() throws Exception {
        syncEventCache = CuratorCache.build(curator, syncEventsPath);
        
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(node -> {
                    try {
                        if (!leaderElection.isLeader() && isMetadataStoreReady) {
                            // Follower 接收到新的同步事件，且元数据存储已准备好
                            processSyncEvent(node.getPath(), node.getData());
                        } else if (!isMetadataStoreReady) {
                            log.debug("元数据存储尚未准备好，跳过同步事件: {}", node.getPath());
                        }
                    } catch (Exception e) {
                        log.error("处理同步事件创建失败", e);
                    }
                })
                .build();
        
        syncEventCache.listenable().addListener(listener);
        syncEventCache.start();
    }
    
    /**
     * Leader 发布同步事件
     */
    public void publishSyncEvent(SyncEventType type, String path, Map<String, Object> data) {
        if (!leaderElection.isLeader()) {
            log.warn("只有 Leader 可以发布同步事件");
            return;
        }
        
        try {
            long sequence = lastSyncSequence.incrementAndGet();
            
            SyncEvent event = new SyncEvent(type, path, data);
            event.sequence = sequence;
            event.leaderId = leaderElection.getCurrentInstance().getInstanceId();
            
            String eventPath = syncEventsPath + "/" + String.format("%020d", sequence);
            String eventData = objectMapper.writeValueAsString(event);
            
            curator.create()
                    .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                    .forPath(eventPath, eventData.getBytes(StandardCharsets.UTF_8));
            
            log.debug("发布同步事件: {} -> {}", sequence, type);
            
        } catch (Exception e) {
            log.error("发布同步事件失败", e);
        }
    }
    
    /**
     * 发布文件创建同步事件
     */
    public void publishFileCreateEvent(String path, Inode inode) {
        Map<String, Object> data = new HashMap<>();
        data.put("fileId", inode.getFileId());
        data.put("name", inode.getName());
        data.put("parentId", inode.getParentId());
        data.put("permissions", inode.getPermissions());
        data.put("owner", inode.getOwner());
        data.put("group", inode.getGroup());
        data.put("replication", inode.getReplication());
        data.put("size", inode.getSize());
        data.put("ctime", inode.getCtime().toString());
        data.put("mtime", inode.getMtime().toString());
        
        publishSyncEvent(SyncEventType.FILE_CREATE, path, data);
    }
    
    /**
     * 发布目录创建同步事件
     */
    public void publishDirectoryCreateEvent(String path, Inode inode) {
        Map<String, Object> data = new HashMap<>();
        data.put("fileId", inode.getFileId());
        data.put("name", inode.getName());
        data.put("parentId", inode.getParentId());
        data.put("permissions", inode.getPermissions());
        data.put("owner", inode.getOwner());
        data.put("group", inode.getGroup());
        data.put("ctime", inode.getCtime().toString());
        data.put("mtime", inode.getMtime().toString());
        
        publishSyncEvent(SyncEventType.DIRECTORY_CREATE, path, data);
    }
    
    /**
     * 发布块分配同步事件
     */
    public void publishBlockAllocateEvent(String fileId, List<BlockInfo> blocks) {
        Map<String, Object> data = new HashMap<>();
        List<Map<String, Object>> blocksData = new ArrayList<>();
        
        for (BlockInfo block : blocks) {
            Map<String, Object> blockData = new HashMap<>();
            blockData.put("blkIdx", block.getBlkIdx());
            blockData.put("size", block.getSize());
            
            List<Map<String, Object>> replicasData = new ArrayList<>();
            for (BlockInfo.ReplicaInfo replica : block.getReplicas()) {
                Map<String, Object> replicaData = new HashMap<>();
                replicaData.put("dsId", replica.getDsId());
                replicaData.put("host", replica.getHost());
                replicaData.put("port", replica.getPort());
                replicaData.put("role", replica.getRole().toString());
                replicasData.add(replicaData);
            }
            blockData.put("replicas", replicasData);
            blocksData.add(blockData);
        }
        
        data.put("blocks", blocksData);
        publishSyncEvent(SyncEventType.BLOCK_ALLOCATE, fileId, data);
    }
    
    /**
     * Follower 处理同步事件
     */
    private void processSyncEvent(String eventPath, byte[] eventData) {
        try {
            String eventJson = new String(eventData, StandardCharsets.UTF_8);
            
            // 检查数据是否为空或只包含空白字符
            if (eventJson.trim().isEmpty()) {
                log.debug("忽略空的同步事件数据: {}", eventPath);
                return;
            }
            
            // 检查是否是有效的JSON格式（必须以{开始）
            if (!eventJson.trim().startsWith("{")) {
                log.warn("跳过非JSON格式的同步事件数据: {} -> {}", eventPath, eventJson.trim());
                // 如果是旧的IP地址格式或其他非JSON数据，忽略它
                return;
            }
            
            // 验证是否看起来像有效的IP地址（简单检查）
            if (isPlainIpAddress(eventJson.trim())) {
                log.warn("跳过纯IP地址格式的同步事件数据: {} -> {}", eventPath, eventJson.trim());
                return;
            }
            
            SyncEvent event = objectMapper.readValue(eventJson, SyncEvent.class);
            
            // 检查事件序列号
            if (event.sequence <= lastSyncSequence.get()) {
                log.debug("忽略已处理的同步事件: {}", event.sequence);
                return;
            }
            
            // 应用同步事件
            applySyncEvent(event);
            
            // 更新同步序列号
            lastSyncSequence.set(event.sequence);
            
            // 更新同步状态
            updateSyncStatus(event.sequence, null);
            
            log.debug("处理同步事件完成: {} -> {}", event.sequence, event.type);
            
        } catch (Exception e) {
            log.error("处理同步事件失败: {}", eventPath, e);
            updateSyncStatus(0, e.getMessage());
        }
    }
    
    /**
     * 检查字符串是否是纯IP地址格式
     */
    private boolean isPlainIpAddress(String data) {
        // 简单的IP地址格式检查：由数字和点组成，且包含3个点
        if (data == null || data.isEmpty()) {
            return false;
        }
        
        // 检查是否符合IP地址的基本格式
        String[] parts = data.split("\\.");
        if (parts.length != 4) {
            return false;
        }
        
        for (String part : parts) {
            try {
                int num = Integer.parseInt(part);
                if (num < 0 || num > 255) {
                    return false;
                }
            } catch (NumberFormatException e) {
                return false;
            }
        }
        
        return true;
    }
    
    /**
     * 应用同步事件到本地存储
     */
    private void applySyncEvent(SyncEvent event) throws Exception {
        switch (event.type) {
            case FILE_CREATE:
                applyFileCreate(event);
                break;
            case FILE_DELETE:
                applyFileDelete(event);
                break;
            case DIRECTORY_CREATE:
                applyDirectoryCreate(event);
                break;
            case DIRECTORY_DELETE:
                applyDirectoryDelete(event);
                break;
            case METADATA_UPDATE:
                applyMetadataUpdate(event);
                break;
            case BLOCK_ALLOCATE:
                applyBlockAllocate(event);
                break;
            case BLOCK_COMMIT:
                applyBlockCommit(event);
                break;
            default:
                log.warn("未知的同步事件类型: {}", event.type);
        }
    }
    
    private void applyFileCreate(SyncEvent event) throws Exception {
        // 实现文件创建的同步逻辑
        String path = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步文件创建: {}", path);
        
        try {
            // 从同步事件数据中重建Inode对象
            String fileId = (String) data.get("fileId");
            String name = (String) data.get("name");
            String parentId = (String) data.get("parentId");
            Integer permissions = (Integer) data.get("permissions");
            String owner = (String) data.get("owner");
            String group = (String) data.get("group");
            Integer replication = (Integer) data.get("replication");
            Long size = ((Number) data.get("size")).longValue();
            String ctimeStr = (String) data.get("ctime");
            String mtimeStr = (String) data.get("mtime");
            
            // 构建文件Inode
            Inode fileInode = Inode.builder()
                    .fileId(fileId)
                    .type(FileType.FILE)
                    .name(name)
                    .parentId(parentId)
                    .ctime(LocalDateTime.parse(ctimeStr))
                    .mtime(LocalDateTime.parse(mtimeStr))
                    .size(size)
                    .replication(replication)
                    .owner(owner)
                    .group(group)
                    .permissions(permissions)
                    .build();
            
            // 保存到本地元数据存储
            metadataStore.putInode(fileInode);
            if (parentId != null) {
                metadataStore.putDentry(parentId, name, fileId);
            }
            metadataStore.putPathCache(path, fileId);
            
            log.info("同步文件创建完成: {} -> {}", path, fileId);
            
        } catch (Exception e) {
            log.error("同步文件创建失败: {}", path, e);
            throw e;
        }
    }
    
    private void applyFileDelete(SyncEvent event) throws Exception {
        // 实现文件删除的同步逻辑
        String path = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步文件删除: {}", path);
        
        try {
            String fileId = (String) data.get("fileId");
            String name = (String) data.get("name");
            String parentId = (String) data.get("parentId");
            
            // 检查文件是否存在
            Optional<Inode> fileInodeOpt = metadataStore.getInode(fileId);
            if (!fileInodeOpt.isPresent()) {
                log.debug("同步删除时文件不存在，可能已被删除: {}", path);
                // 仍然需要清理路径缓存，防止缓存残留
                metadataStore.deletePathCache(path);
                return;
            }
            
            // 删除文件的所有块
            List<BlockInfo> blocks = metadataStore.getFileBlocks(fileId);
            for (BlockInfo block : blocks) {
                metadataStore.deleteBlock(fileId, block.getBlkIdx());
            }
            
            // 删除inode
            metadataStore.deleteInode(fileId);
            
            // 删除目录项
            if (parentId != null && name != null) {
                metadataStore.deleteDentry(parentId, name);
            }
            
            // 删除路径缓存
            metadataStore.deletePathCache(path);
            
            log.info("同步文件删除完成: {}", path);
            
        } catch (Exception e) {
            log.error("同步文件删除失败: {}", path, e);
            throw e;
        }
    }
    
    private void applyDirectoryCreate(SyncEvent event) throws Exception {
        // 实现目录创建的同步逻辑
        String path = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步目录创建: {}", path);
        
        try {
            // 从同步事件数据中重建Inode对象
            String fileId = (String) data.get("fileId");
            String name = (String) data.get("name");
            String parentId = (String) data.get("parentId");
            Integer permissions = (Integer) data.get("permissions");
            String owner = (String) data.get("owner");
            String group = (String) data.get("group");
            String ctimeStr = (String) data.get("ctime");
            String mtimeStr = (String) data.get("mtime");
            
            // 构建目录Inode
            Inode dirInode = Inode.builder()
                    .fileId(fileId)
                    .type(FileType.DIR)
                    .name(name)
                    .parentId(parentId)
                    .ctime(LocalDateTime.parse(ctimeStr))
                    .mtime(LocalDateTime.parse(mtimeStr))
                    .size(0)
                    .replication(0)
                    .owner(owner)
                    .group(group)
                    .permissions(permissions)
                    .build();
            
            // 保存到本地元数据存储
            metadataStore.putInode(dirInode);
            if (parentId != null) {
                metadataStore.putDentry(parentId, name, fileId);
            }
            metadataStore.putPathCache(path, fileId);
            
            log.info("同步目录创建完成: {} -> {}", path, fileId);
            
        } catch (Exception e) {
            log.error("同步目录创建失败: {}", path, e);
            throw e;
        }
    }
    
    private void applyDirectoryDelete(SyncEvent event) throws Exception {
        // 实现目录删除的同步逻辑
        String path = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步目录删除: {}", path);
        
        try {
            String fileId = (String) data.get("fileId");
            String name = (String) data.get("name");
            String parentId = (String) data.get("parentId");
            
            // 检查目录是否存在
            Optional<Inode> dirInodeOpt = metadataStore.getInode(fileId);
            if (!dirInodeOpt.isPresent()) {
                log.debug("同步删除时目录不存在，可能已被删除: {}", path);
                // 仍然需要清理路径缓存，防止缓存残留
                metadataStore.deletePathCache(path);
                return;
            }
            
            // 只删除空目录（子项应该已经通过单独的删除事件被处理了）
            syncDeleteEmptyDirectory(fileId, name, parentId, path);
            
            log.info("同步目录删除完成: {}", path);
            
        } catch (Exception e) {
            log.error("同步目录删除失败: {}", path, e);
            throw e;
        }
    }
    

    
    /**
     * 同步删除空目录
     */
    private void syncDeleteEmptyDirectory(String dirId, String name, String parentId, String path) throws Exception {
        // 检查目录是否仍然存在
        Optional<Inode> dirInodeOpt = metadataStore.getInode(dirId);
        if (!dirInodeOpt.isPresent()) {
            log.debug("目录已不存在，跳过删除: {}", path);
            // 仍然需要清理路径缓存
            metadataStore.deletePathCache(path);
            return;
        }
        
        // 获取目录下的子项列表
        List<String> children = metadataStore.listDentries(dirId);
        
        // 如果目录不为空，强制清理剩余的子项
        if (!children.isEmpty()) {
            log.warn("目录不为空，强制清理剩余子项: {} ({}个子项)", path, children.size());
            for (String childName : children) {
                try {
                    metadataStore.deleteDentry(dirId, childName);
                    // 清理子项的路径缓存
                    String childPath = path.endsWith("/") ? path + childName : path + "/" + childName;
                    metadataStore.deletePathCache(childPath);
                } catch (Exception e) {
                    log.warn("清理子项失败: {}/{}", path, childName, e);
                }
            }
        }
        
        // 删除inode
        metadataStore.deleteInode(dirId);
        
        // 删除父目录中的目录项
        if (parentId != null && name != null) {
            metadataStore.deleteDentry(parentId, name);
        }
        
        // 删除路径缓存
        metadataStore.deletePathCache(path);
        
        log.debug("空目录删除完成: {}", path);
    }
    

    
    private void applyMetadataUpdate(SyncEvent event) throws Exception {
        // 实现元数据更新的同步逻辑
        String path = event.path;
        // Map<String, Object> metadata = event.data; // TODO: 实现元数据更新逻辑
        
        log.debug("同步元数据更新: {}", path);
        // metadataStore.updateMetadata(path, metadata);
    }
    
    private void applyBlockAllocate(SyncEvent event) throws Exception {
        // 实现块分配的同步逻辑
        String fileId = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步块分配: fileId={}", fileId);
        
        try {
            // 从同步事件数据中提取块分配信息
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> blocksData = (List<Map<String, Object>>) data.get("blocks");
            
            if (blocksData != null) {
                for (Map<String, Object> blockData : blocksData) {
                    // 重建BlockInfo对象
                    Integer blkIdx = (Integer) blockData.get("blkIdx");
                    Long size = ((Number) blockData.get("size")).longValue();
                    
                    @SuppressWarnings("unchecked")
                    List<Map<String, Object>> replicasData = (List<Map<String, Object>>) blockData.get("replicas");
                    
                    List<BlockInfo.ReplicaInfo> replicas = new ArrayList<>();
                    if (replicasData != null) {
                        for (Map<String, Object> replicaData : replicasData) {
                            String dsId = (String) replicaData.get("dsId");
                            String host = (String) replicaData.get("host");
                            Integer port = (Integer) replicaData.get("port");
                            String roleStr = (String) replicaData.get("role");
                            
                            BlockInfo.ReplicaInfo.ReplicaRole role = 
                                BlockInfo.ReplicaInfo.ReplicaRole.valueOf(roleStr);
                            
                            BlockInfo.ReplicaInfo replica = BlockInfo.ReplicaInfo.builder()
                                    .dsId(dsId)
                                    .host(host)
                                    .port(port)
                                    .role(role)
                                    .build();
                            
                            replicas.add(replica);
                        }
                    }
                    
                    // 创建BlockInfo对象
                    BlockInfo blockInfo = BlockInfo.builder()
                            .fileId(fileId)
                            .blkIdx(blkIdx)
                            .size(size)
                            .replicas(replicas)
                            .build();
                    
                    // 保存到本地元数据存储
                    metadataStore.putBlock(fileId, blkIdx, blockInfo);
                    
                    log.debug("同步块分配完成: fileId={}, blkIdx={}, size={}, replicas={}", 
                            fileId, blkIdx, size, replicas.size());
                }
            }
            
            log.info("同步块分配完成: fileId={}, 共{}个块", fileId, 
                    blocksData != null ? blocksData.size() : 0);
            
        } catch (Exception e) {
            log.error("同步块分配失败: fileId={}", fileId, e);
            throw e;
        }
    }
    
    private void applyBlockCommit(SyncEvent event) throws Exception {
        // 实现块提交的同步逻辑
        String fileId = event.path;
        Map<String, Object> data = event.data;
        
        log.debug("同步块提交: fileId={}", fileId);
        
        try {
            // 从同步事件数据中提取文件信息
            Long size = ((Number) data.get("size")).longValue();
            String checksum = (String) data.get("checksum");
            String mtimeStr = (String) data.get("mtime");
            LocalDateTime mtime = LocalDateTime.parse(mtimeStr);
            
            // 获取并更新文件inode
            Optional<Inode> inodeOpt = metadataStore.getInode(fileId);
            if (inodeOpt.isPresent()) {
                Inode fileInode = inodeOpt.get();
                fileInode.setSize(size);
                fileInode.setMtime(mtime);
                if (checksum != null) {
                    fileInode.setChecksum(checksum);
                }
                
                // 保存更新后的inode
                metadataStore.putInode(fileInode);
                log.debug("同步文件inode更新完成: fileId={}, size={}", fileId, size);
            } else {
                log.warn("同步块提交时未找到文件inode: fileId={}", fileId);
            }
            
            // 从同步事件数据中提取块信息并更新
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> blocksData = (List<Map<String, Object>>) data.get("blocks");
            
            if (blocksData != null) {
                for (Map<String, Object> blockData : blocksData) {
                    // 重建BlockInfo对象
                    Integer blkIdx = (Integer) blockData.get("blkIdx");
                    Long blockSize = ((Number) blockData.get("size")).longValue();
                    
                    @SuppressWarnings("unchecked")
                    List<Map<String, Object>> replicasData = (List<Map<String, Object>>) blockData.get("replicas");
                    
                    List<BlockInfo.ReplicaInfo> replicas = new ArrayList<>();
                    if (replicasData != null) {
                        for (Map<String, Object> replicaData : replicasData) {
                            String dsId = (String) replicaData.get("dsId");
                            String host = (String) replicaData.get("host");
                            Integer port = (Integer) replicaData.get("port");
                            String roleStr = (String) replicaData.get("role");
                            
                            BlockInfo.ReplicaInfo.ReplicaRole role = 
                                BlockInfo.ReplicaInfo.ReplicaRole.valueOf(roleStr);
                            
                            BlockInfo.ReplicaInfo replica = BlockInfo.ReplicaInfo.builder()
                                    .dsId(dsId)
                                    .host(host)
                                    .port(port)
                                    .role(role)
                                    .build();
                            
                            replicas.add(replica);
                        }
                    }
                    
                    // 创建BlockInfo对象
                    BlockInfo blockInfo = BlockInfo.builder()
                            .fileId(fileId)
                            .blkIdx(blkIdx)
                            .size(blockSize)
                            .replicas(replicas)
                            .build();
                    
                    // 更新块信息到本地元数据存储
                    metadataStore.putBlock(fileId, blkIdx, blockInfo);
                    
                    log.debug("同步块提交完成: fileId={}, blkIdx={}, size={}", 
                            fileId, blkIdx, blockSize);
                }
            }
            
            log.info("同步块提交完成: fileId={}, size={}, 共{}个块", fileId, size,
                    blocksData != null ? blocksData.size() : 0);
            
        } catch (Exception e) {
            log.error("同步块提交失败: fileId={}", fileId, e);
            throw e;
        }
    }
    
    private void checkFollowerSyncStatus() {
        // Leader 检查所有 Follower 的同步状态
        try {
            List<String> followers = curator.getChildren().forPath(syncStatusPath);
            
            for (String follower : followers) {
                String statusPath = syncStatusPath + "/" + follower;
                byte[] statusData = curator.getData().forPath(statusPath);
                
                if (statusData != null) {
                    String statusJson = new String(statusData, StandardCharsets.UTF_8);
                    SyncStatus status = objectMapper.readValue(statusJson, SyncStatus.class);
                    
                    followerSyncStatus.put(follower, status);
                    
                    // 检查同步延迟
                    long lag = lastSyncSequence.get() - status.lastSyncSequence;
                    if (lag > 100) { // 如果落后超过100个事件
                        log.warn("Follower {} 同步延迟: {} 个事件", follower, lag);
                    }
                }
            }
        } catch (Exception e) {
            log.error("检查 Follower 同步状态失败", e);
        }
    }
    
    private void pullSyncEventsFromLeader() {
        try {
            MetaServerInfo leader = leaderElection.getCurrentLeader();
            if (leader == null) {
                return;
            }
            
            // 构造 Leader 的同步API URL
            String leaderUrl = String.format("http://%s:%d/admin/sync/events?from=%d", 
                    leader.getHost(), leader.getPort(), lastSyncSequence.get() + 1);
            
            ResponseEntity<SyncEvent[]> response = restTemplate.getForEntity(leaderUrl, SyncEvent[].class);
            
            if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
                SyncEvent[] events = response.getBody();
                
                for (SyncEvent event : events) {
                    applySyncEvent(event);
                    lastSyncSequence.set(Math.max(lastSyncSequence.get(), event.sequence));
                }
                
                if (events.length > 0) {
                    log.debug("从 Leader 同步了 {} 个事件", events.length);
                    updateSyncStatus(lastSyncSequence.get(), null);
                }
            }
            
        } catch (Exception e) {
            log.error("从 Leader 拉取同步事件失败", e);
            updateSyncStatus(0, e.getMessage());
        }
    }
    
    private void updateSyncStatus(long sequence, String error) {
        try {
            String instanceId = leaderElection.getCurrentInstance().getInstanceId();
            
            SyncStatus status = new SyncStatus(instanceId);
            status.lastSyncSequence = sequence > 0 ? sequence : lastSyncSequence.get();
            status.lastSyncTime = LocalDateTime.now();
            status.syncError = error;
            
            String statusPath = syncStatusPath + "/" + instanceId;
            String statusData = objectMapper.writeValueAsString(status);
            
            if (curator.checkExists().forPath(statusPath) != null) {
                curator.setData().forPath(statusPath, statusData.getBytes(StandardCharsets.UTF_8));
            } else {
                curator.create()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(statusPath, statusData.getBytes(StandardCharsets.UTF_8));
            }
            
        } catch (Exception e) {
            log.error("更新同步状态失败", e);
        }
    }
    
    /**
     * 清理无效的同步事件数据
     */
    private void cleanupInvalidSyncEvents() {
        try {
            if (curator.checkExists().forPath(syncEventsPath) == null) {
                log.debug("同步事件路径不存在，跳过清理: {}", syncEventsPath);
                return;
            }
            
            List<String> events = curator.getChildren().forPath(syncEventsPath);
            int cleanedCount = 0;
            
            for (String event : events) {
                String eventPath = syncEventsPath + "/" + event;
                try {
                    byte[] eventData = curator.getData().forPath(eventPath);
                    
                    if (eventData == null || eventData.length == 0) {
                        curator.delete().forPath(eventPath);
                        cleanedCount++;
                        log.debug("清理空数据的同步事件: {}", event);
                        continue;
                    }
                    
                    String eventJson = new String(eventData, StandardCharsets.UTF_8).trim();
                    
                    // 检查是否是无效格式（如纯IP地址）
                    if (eventJson.isEmpty() || !eventJson.startsWith("{") || isPlainIpAddress(eventJson)) {
                        curator.delete().forPath(eventPath);
                        cleanedCount++;
                        log.info("清理无效格式的同步事件: {} -> {}", event, eventJson);
                        continue;
                    }
                    
                    // 尝试解析JSON，如果失败则删除
                    try {
                        objectMapper.readValue(eventJson, SyncEvent.class);
                    } catch (Exception jsonError) {
                        curator.delete().forPath(eventPath);
                        cleanedCount++;
                        log.info("清理无法解析JSON的同步事件: {} -> {}", event, eventJson);
                    }
                    
                } catch (Exception e) {
                    log.warn("处理同步事件清理时出错: {} -> {}", event, e.getMessage());
                }
            }
            
            if (cleanedCount > 0) {
                log.info("清理了 {} 个无效的同步事件", cleanedCount);
            }
            
        } catch (Exception e) {
            log.error("清理无效同步事件失败", e);
        }
    }
    
    private void cleanupOldSyncEvents() {
        try {
            // 清理超过1小时的同步事件
            long cutoffTime = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1);
            
            List<String> events = curator.getChildren().forPath(syncEventsPath);
            
            for (String event : events) {
                String eventPath = syncEventsPath + "/" + event;
                byte[] eventData = curator.getData().forPath(eventPath);
                
                if (eventData != null) {
                    String eventJson = new String(eventData, StandardCharsets.UTF_8);
                    
                    // 跳过无效格式的数据
                    if (!eventJson.trim().startsWith("{")) {
                        continue;
                    }
                    
                    try {
                        SyncEvent syncEvent = objectMapper.readValue(eventJson, SyncEvent.class);
                        
                        if (syncEvent.timestamp < cutoffTime) {
                            curator.delete().forPath(eventPath);
                            log.debug("清理过期同步事件: {}", event);
                        }
                    } catch (Exception e) {
                        log.debug("跳过无法解析的同步事件: {} -> {}", event, eventJson);
                    }
                }
            }
        } catch (Exception e) {
            log.error("清理过期同步事件失败", e);
        }
    }
    
    /**
     * 获取当前同步状态
     */
    public Map<String, Object> getCurrentSyncStatus() {
        String instanceId = leaderElection.getCurrentInstance().getInstanceId();
        Map<String, Object> status = new HashMap<>();
        status.put("instanceId", instanceId);
        status.put("lastSyncSequence", lastSyncSequence.get());
        status.put("lastSyncTime", LocalDateTime.now().toString());
        status.put("isMetadataStoreReady", isMetadataStoreReady);
        status.put("isInitialized", isInitialized);
        return status;
    }
    
    /**
     * 获取所有Follower的同步状态
     */
    public Map<String, SyncStatus> getAllFollowerSyncStatus() {
        return new HashMap<>(followerSyncStatus);
    }
    
    /**
     * 执行全集群数据一致性检查
     * 比较主节点和所有从节点的数据，检测不一致情况
     */
    public ConsistencyCheckResult performConsistencyCheck() {
        if (!leaderElection.isLeader()) {
            throw new IllegalStateException("只有Leader节点可以执行一致性检查");
        }
        
        log.info("开始执行全集群数据一致性检查");
        ConsistencyCheckResult result = new ConsistencyCheckResult();
        
        try {
            // 获取Leader节点的数据摘要
            DataSummary leaderSummary = getLocalDataSummary();
            log.info("Leader数据摘要: inodes={}, blocks={}, checksum={}", 
                    leaderSummary.totalInodes, leaderSummary.totalBlocks, leaderSummary.checksum);
            
            // 检查每个Follower节点
            List<String> followers = curator.getChildren().forPath(syncStatusPath);
            boolean allConsistent = true;
            
            for (String followerId : followers) {
                try {
                    DataSummary followerSummary = getFollowerDataSummary(followerId);
                    if (followerSummary == null) {
                        log.warn("无法获取Follower {}的数据摘要", followerId);
                        continue;
                    }
                    
                    log.info("Follower {} 数据摘要: inodes={}, blocks={}, checksum={}", 
                            followerId, followerSummary.totalInodes, followerSummary.totalBlocks, followerSummary.checksum);
                    
                    // 检查数据一致性
                    List<InconsistencyInfo> inconsistencies = compareDataSummaries(leaderSummary, followerSummary, followerId);
                    
                    if (!inconsistencies.isEmpty()) {
                        allConsistent = false;
                        for (InconsistencyInfo inconsistency : inconsistencies) {
                            result.inconsistencies.put(followerId + "_" + inconsistency.type, inconsistency);
                        }
                        log.warn("检测到Follower {}存在{}个数据不一致问题", followerId, inconsistencies.size());
                    } else {
                        log.info("Follower {}数据一致性检查通过", followerId);
                    }
                    
                } catch (Exception e) {
                    log.error("检查Follower {}时出错", followerId, e);
                    InconsistencyInfo error = new InconsistencyInfo("CHECK_ERROR", 
                            "无法检查节点数据: " + e.getMessage(), "检查节点连接状态");
                    result.inconsistencies.put(followerId + "_ERROR", error);
                    allConsistent = false;
                }
            }
            
            result.isConsistent = allConsistent;
            
            if (allConsistent) {
                log.info("数据一致性检查完成：所有节点数据一致");
            } else {
                log.warn("数据一致性检查完成：发现{}个不一致问题", result.inconsistencies.size());
            }
            
        } catch (Exception e) {
            log.error("执行一致性检查时出错", e);
            result.isConsistent = false;
            InconsistencyInfo error = new InconsistencyInfo("SYSTEM_ERROR", 
                    "一致性检查系统错误: " + e.getMessage(), "检查系统状态和日志");
            result.inconsistencies.put("SYSTEM_ERROR", error);
        }
        
        return result;
    }
    
    /**
     * 获取本地数据摘要
     */
    private DataSummary getLocalDataSummary() throws Exception {
        DataSummary summary = new DataSummary();
        
        // 统计Inode数量
        summary.totalInodes = metadataStore.getTotalInodeCount();
        
        // 统计Block数量
        summary.totalBlocks = metadataStore.getTotalBlockCount();
        
        // 计算数据校验和（基于关键元数据）
        summary.checksum = calculateDataChecksum();
        
        return summary;
    }
    
    /**
     * 获取Follower节点的数据摘要
     */
    private DataSummary getFollowerDataSummary(String followerId) {
        try {
            MetaServerInfo follower = getFollowerInfo(followerId);
            if (follower == null) {
                return null;
            }
            
            String url = String.format("http://%s:%d/admin/data-summary", 
                    follower.getHost(), follower.getPort());
            
            ResponseEntity<DataSummary> response = restTemplate.getForEntity(url, DataSummary.class);
            
            if (response.getStatusCode().is2xxSuccessful()) {
                return response.getBody();
            }
            
        } catch (Exception e) {
            log.error("获取Follower {}数据摘要失败", followerId, e);
        }
        
        return null;
    }
    
    /**
     * 比较Leader和Follower的数据摘要
     */
    private List<InconsistencyInfo> compareDataSummaries(DataSummary leader, DataSummary follower, String followerId) {
        List<InconsistencyInfo> inconsistencies = new ArrayList<>();
        
        // 检查Inode数量
        if (follower.totalInodes > leader.totalInodes) {
            InconsistencyInfo info = new InconsistencyInfo("EXTRA_INODES", 
                    String.format("Follower节点有多余的Inode：%d > %d", follower.totalInodes, leader.totalInodes),
                    "执行从节点数据重置");
            info.details.put("followerCount", follower.totalInodes);
            info.details.put("leaderCount", leader.totalInodes);
            info.details.put("excess", follower.totalInodes - leader.totalInodes);
            inconsistencies.add(info);
        } else if (follower.totalInodes < leader.totalInodes) {
            InconsistencyInfo info = new InconsistencyInfo("MISSING_INODES", 
                    String.format("Follower节点缺少Inode：%d < %d", follower.totalInodes, leader.totalInodes),
                    "执行增量同步");
            info.details.put("followerCount", follower.totalInodes);
            info.details.put("leaderCount", leader.totalInodes);
            info.details.put("missing", leader.totalInodes - follower.totalInodes);
            inconsistencies.add(info);
        }
        
        // 检查Block数量
        if (follower.totalBlocks > leader.totalBlocks) {
            InconsistencyInfo info = new InconsistencyInfo("EXTRA_BLOCKS", 
                    String.format("Follower节点有多余的Block：%d > %d", follower.totalBlocks, leader.totalBlocks),
                    "执行从节点数据重置");
            info.details.put("followerCount", follower.totalBlocks);
            info.details.put("leaderCount", leader.totalBlocks);
            info.details.put("excess", follower.totalBlocks - leader.totalBlocks);
            inconsistencies.add(info);
        } else if (follower.totalBlocks < leader.totalBlocks) {
            InconsistencyInfo info = new InconsistencyInfo("MISSING_BLOCKS", 
                    String.format("Follower节点缺少Block：%d < %d", follower.totalBlocks, leader.totalBlocks),
                    "执行增量同步");
            info.details.put("followerCount", follower.totalBlocks);
            info.details.put("leaderCount", leader.totalBlocks);
            info.details.put("missing", leader.totalBlocks - follower.totalBlocks);
            inconsistencies.add(info);
        }
        
        // 检查数据校验和
        if (!leader.checksum.equals(follower.checksum)) {
            InconsistencyInfo info = new InconsistencyInfo("DATA_MISMATCH", 
                    "数据内容不匹配（校验和不同）",
                    "执行全量数据重置");
            info.details.put("leaderChecksum", leader.checksum);
            info.details.put("followerChecksum", follower.checksum);
            inconsistencies.add(info);
        }
        
        return inconsistencies;
    }
    
    /**
     * 计算本地数据的校验和
     */
    private String calculateDataChecksum() throws Exception {
        // 这里使用一个简化的校验和计算方法
        // 在实际生产环境中，可能需要更复杂的校验和算法
        java.security.MessageDigest md = java.security.MessageDigest.getInstance("MD5");
        
        // 添加关键统计信息到校验和计算中
        md.update(String.valueOf(metadataStore.getTotalInodeCount()).getBytes());
        md.update(String.valueOf(metadataStore.getTotalBlockCount()).getBytes());
        
        // 可以添加更多关键数据，如根目录的最后修改时间等
        Optional<Inode> rootInode = metadataStore.getInode("root");
        if (rootInode.isPresent()) {
            md.update(rootInode.get().getMtime().toString().getBytes());
        }
        
        byte[] digest = md.digest();
        StringBuilder sb = new StringBuilder();
        for (byte b : digest) {
            sb.append(String.format("%02x", b));
        }
        
        return sb.toString();
    }
    
    /**
     * 获取Follower节点信息
     */
    private MetaServerInfo getFollowerInfo(String followerId) throws Exception {
        // 从ZooKeeper获取Follower节点信息
        String metaServersPath = properties.getZookeeper().getRootPath() + "/metaservers";
        List<String> children = curator.getChildren().forPath(metaServersPath);
        
        for (String child : children) {
            if (child.equals(followerId)) {
                String nodePath = metaServersPath + "/" + child;
                byte[] nodeData = curator.getData().forPath(nodePath);
                if (nodeData != null) {
                    String nodeJson = new String(nodeData, StandardCharsets.UTF_8);
                    return objectMapper.readValue(nodeJson, MetaServerInfo.class);
                }
            }
        }
        
        return null;
    }
    
    /**
     * 执行从节点数据重置
     * 当从节点数据比主节点多时，清空从节点数据并重新同步
     */
    public boolean resetFollowerData(String followerId) {
        if (!leaderElection.isLeader()) {
            throw new IllegalStateException("只有Leader节点可以执行从节点重置");
        }
        
        log.info("开始重置Follower节点数据: {}", followerId);
        
        try {
            MetaServerInfo follower = getFollowerInfo(followerId);
            if (follower == null) {
                log.error("找不到Follower节点信息: {}", followerId);
                return false;
            }
            
            // 调用Follower节点的重置API
            String resetUrl = String.format("http://%s:%d/admin/reset-data", 
                    follower.getHost(), follower.getPort());
            
            ResponseEntity<String> response = restTemplate.postForEntity(resetUrl, null, String.class);
            
            if (response.getStatusCode().is2xxSuccessful()) {
                log.info("Follower节点{}数据重置成功", followerId);
                
                // 触发全量数据同步
                triggerFullDataSync(followerId);
                
                return true;
            } else {
                log.error("Follower节点{}数据重置失败: {}", followerId, response.getStatusCode());
                return false;
            }
            
        } catch (Exception e) {
            log.error("重置Follower节点{}数据时出错", followerId, e);
            return false;
        }
    }
    
    /**
     * 触发全量数据同步
     */
    private void triggerFullDataSync(String followerId) {
        log.info("开始为Follower节点{}执行全量数据同步", followerId);
        
        syncExecutor.submit(() -> {
            try {
                // 获取所有Inode并发布同步事件
                List<Inode> allInodes = metadataStore.getAllInodes();
                
                for (Inode inode : allInodes) {
                    String path = metadataStore.getInodePath(inode.getFileId());
                    if (path != null) {
                        if (inode.isDirectory()) {
                            publishDirectoryCreateEvent(path, inode);
                        } else {
                            publishFileCreateEvent(path, inode);
                            
                            // 同步文件的块信息
                            List<BlockInfo> blocks = metadataStore.getFileBlocks(inode.getFileId());
                            if (!blocks.isEmpty()) {
                                publishBlockAllocateEvent(inode.getFileId(), blocks);
                            }
                        }
                    }
                }
                
                log.info("Follower节点{}全量数据同步完成，共同步{}个Inode", followerId, allInodes.size());
                
            } catch (Exception e) {
                log.error("为Follower节点{}执行全量数据同步时出错", followerId, e);
            }
        });
    }
    
    // 数据摘要类
    public static class DataSummary {
        public long totalInodes;
        public long totalBlocks;
        public String checksum;
        
        public DataSummary() {}
    }
}
